
#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <iostream>
#include <vector>
#include <string>
#include <unordered_map>
#include <cstring>
#include <cassert>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <memory>
#include <typeinfo>
#include <time.h>
#include <unistd.h>
#include <signal.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <arpa/inet.h>


using namespace std;

// 日志宏
#define INF 0
#define DBG 1
#define ERR 1
#define LOG_LEVEL INF
#define LOG(level, format, ...) do{\
    if(level < LOG_LEVEL) break;\
    time_t t = time(NULL);\
    struct tm * ltm = localtime(&t);\
    char buff[64];\
    strftime(buff, sizeof(buff)-1, "%H:%M:%S",ltm);\
    fprintf(stdout, "[%p %s %s:%d]#" format "\n", (void*)pthread_self(), buff, __FILE__, __LINE__, ##__VA_ARGS__);\
}while(0)
#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024

// 缓冲区模块
class Buffer
{
private:
    std::vector<char> _buffer;   // 使用vector进行内存空间管理
    uint64_t _reader_idx;   // 读偏移
    uint64_t _writer_idx;   // 写偏移
public:
    Buffer():_reader_idx(0), _writer_idx(0), _buffer(BUFFER_DEFAULT_SIZE) {}
    char* Begin() { return &*_buffer.begin(); }
    // 获取当前写入起始地址,buffer的起始地址+偏移量
    char* WritePosition() { return Begin() + _writer_idx; }
    // 获取当前读取起始地址
    char* ReadPosition() { return Begin() + _reader_idx; }
    // 获取缓冲区末尾空闲空间大小 ---写偏移之后的空闲空间
    uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }
    // 获取缓冲区开头空闲空间大小 ---读偏移之前的空闲空间
    uint64_t HeadIdleSize() { return _reader_idx; }
    // 获取可读数据大小
    uint64_t ReadableSize() { return _writer_idx - _reader_idx; }
    // 将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        assert(len <= ReadableSize());  //向后移动的大小，必须小于可读数据的大小
        _reader_idx += len;
    }
    // 将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailIdleSize());  //向后移动的大小，必须小于缓冲区末尾空闲空间大小
        _writer_idx += len;
    }
    // 确保可写空间足够 ----(整体空闲空间够了就移动数据，否则就扩容)
    void EnsureWriteSpace(uint64_t len)
    {
        if(len <= TailIdleSize()) //空间足够直接返回
            return;
        // 末尾空间不够，但是整体空间足够，挪动数据腾出空间
        if(len <= TailIdleSize() + HeadIdleSize())  
        {
            uint64_t rsz = ReadableSize();
            std::copy(ReadPosition(), ReadPosition()+rsz, Begin());
            // 更新偏移位置
            _reader_idx = 0;
            _writer_idx = rsz;
        }
        else    //整体空间也不够，扩容
        {
            _buffer.resize(_writer_idx + len);
        }
    }
    // 写入数据
    void Write(const void* data, uint64_t len)
    {
        if(len == 0) return;
        // 保证有足够的空间可以写入
        EnsureWriteSpace(len);
        // 将数据拷贝进缓冲区中
        const char* d = (const char*)data;
        std::copy(d, d+len, WritePosition());
    }
    void WriteAndPush(const void* data, uint64_t len)
    {
        Write(data,len);
        MoveWriteOffset(len);
    }
    void WriteString(const std::string& data)
    {
        return Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string& data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());

    }
    void WriteBuffer(Buffer& data)  //const
    {
        return Write(data.ReadPosition(), data.ReadableSize());
    }
    void WriteBufferAndPush(Buffer& data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadableSize());
    }
    // 读取数据
    void Read(void* buf, uint64_t len)
    {
        // 保证要读取的数据小于可读取的数据
        assert(len <= ReadableSize());
        // 将数据拷贝到buf中
        std::copy(ReadPosition(), ReadPosition()+len, (char*)buf);
    }
    void ReadAndPop(void* buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }
    std::string ReadAsString(uint64_t len)
    {
        // 保证要读取的数据小于可读取的数据
        assert(len <= ReadableSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        assert(len <= ReadableSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    // 查找换行字符的位置
    char* FindCRLF()
    {
        void* res = memchr(ReadPosition(), '\n', ReadableSize());
        return (char*)res;
    }
    // 读取一行数据
    std::string GetLine()
    {
        char* pos = FindCRLF();
        if(nullptr == pos)
            return "";
        return ReadAsString(pos - ReadPosition() + 1);
    }
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    // 清空数据
    void Clear()
    {
        // 不需要清除数据，只需要将偏移量归0即可
        _reader_idx = 0;
        _writer_idx = 0;
    }
};

// 套接字模块
const static int MAX_LISTEN = 1024;
class Socket
{
private:
    int _sockfd;
public:
    Socket():_sockfd(-1) {}
    Socket(int sockfd):_sockfd(sockfd) {}
    ~Socket() { Close(); }
    int Fd() { return _sockfd; }
    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if(_sockfd < 0)
        {
            ERR_LOG("create socket failed");
            return false;
        }
        return true;
    }
    // 绑定地址信息
    bool Bind(const string& ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = bind(_sockfd, (struct sockaddr*)&addr,len);
        if(ret < 0)
        {
            ERR_LOG("bind address failed");
            return false;
        }
        return true;
    }
    // 开始监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        int ret = listen(_sockfd, backlog);
        if(ret < 0)
        {
            ERR_LOG("socket listen failed");
            return false;
        }
        return true;
    }
    // 向服务器发起连接
    bool Connect(const string& ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            ERR_LOG("connect server failed");
            return false;
        }
        return true;
    }
    // 获取新链接
    int Accept()
    {
        int fd = accept(_sockfd, nullptr, nullptr);
        if(fd < 0)
        {
            ERR_LOG("socket accept failed");
            return -1;
        }
        return fd;
    }
    // 接收数据
    ssize_t Recv(void* buf, size_t len, int flag = 0)
    {
        ssize_t ret = recv(_sockfd, buf, len, flag);

        if(ret <= 0)
        {
            // EAGAIN 当前socket的接收缓冲区中没有数据了，只有在非阻塞的情况下才有这个错误
            // EINTR  当前socket的阻塞等待，被信号打断了
            if(errno == EAGAIN || errno == EINTR)
                return 0; //表示没有接收到数据
            ERR_LOG("socket recv failed");
            return -1;
        }
        return ret; //实际接收的数据长度
    }
    ssize_t NonBlockRecv(void* buf, size_t len)
    {
        return Recv(buf,len, MSG_DONTWAIT); //MSG_DONTWAIT表示当前接收为非阻塞
    }
    // 发送数据
    ssize_t Send(const void* buf, size_t len, int flag = 0)
    {
        ssize_t ret = send(_sockfd, buf, len, flag);
        if(ret < 0)
        {
            if(errno == EAGAIN || errno == EINTR)
                return 0; 
            ERR_LOG("socket send failed");
            return -1;
        }
        return ret;//实际发送的数据长度
    }
    ssize_t NonBlockSend(void* buf, size_t len)
    {
        return Send(buf,len, MSG_DONTWAIT); //MSG_DONTWAIT表示当前发送为非阻塞
    }
    // 关闭套接字
    void Close()
    {
        if(_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }
    // 创建一个服务端连接
    bool CreateServer(uint16_t port, const string& ip = "0.0.0.0", bool block_flag = false)
    {
        // 1.创建套接字2.绑定端口3.开始监听4.设置非阻塞5.启用地址重用
        if(Create() == false) return false;
        if(block_flag) NonBlock();
        if(Bind(ip, port) == false) return false;
        if(Listen() == false) return false;
        ReuseAddress();
        return true; 
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t port, const string& ip)
    {
        // 1.创建套接字2.建立连接
        if(Create() == false) return false;
        if(Connect(ip, port) == false) return false;

        return false;
    }
    // 设置套接字选项----开启地址端口重用
    void ReuseAddress()
    {
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }
    // 设置套接字属性----设置为非阻塞
    void NonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag|O_NONBLOCK);
    }
};

// 事件监控
class EventLoop;
class Poller;
class Channel
{
private:
    int _fd;
    uint32_t _events;       //当前需要监控的事件
    uint32_t _revents;      //当前连接触发的事件
    EventLoop* _loop;
    using EventCallback = function<void()>;
    EventCallback _read_callback;           // 可读事件被触发的回调函数
    EventCallback _write_callback;          // 可写事件被触发的回调函数
    EventCallback _error_callback;          // 错误事件被触发的回调函数
    EventCallback _close_callback;          // 连接断开事件被触发的回调函数
    EventCallback _event_callback;          // 任意事件被触发的回调函数
public:
    Channel(EventLoop* loop, int fd):_fd(fd), _events(0), _revents(0), _loop(loop){}
    int Fd() { return _fd; }
    uint32_t Events() { return _events; }   //获取想要监控的事件
    void SetRevents(uint32_t events) { _revents = events; } //设置实际就绪的事件
    void SetReadCallback(const EventCallback& cb)
        {   _read_callback = cb; }
    void SetWriteCallback(const EventCallback& cb)
        {   _write_callback = cb; }
    void SetErrorCallback(const EventCallback& cb)
        {   _error_callback = cb; }
    void SetCloseCallback(const EventCallback& cb)
        {   _close_callback = cb; }
    void SetEventCallback(const EventCallback& cb)
        {   _event_callback = cb; }
    
    bool ReadAble()        // 当前是否监控可读
    {   
        return (_events & EPOLLIN); 
    }
    bool writeAble()       // 当前是否监控可写
    {
        return (_events & EPOLLOUT);
    }
    void EnableRead()      // 启动读时间监控
        {   _events |= EPOLLIN; Update();}   // 后边会添加到eventloop 的事件监控当中
    void Enablewrite()     // 启动写事件监控
        {   _events |= EPOLLOUT; Update();}
    void DisableRead()     // 关闭读事件监控
        {   _events &= ~EPOLLIN; Update();}
    void DisableWrite()    // 关闭写事件监控
        {   _events &= ~EPOLLOUT; Update();}
    void DisableAll()      // 关闭所有事件监控
        {   _events = 0; Update();}
    void Remove();          // 移除监控
    void Update();
    void HandlerEvent()    // 事件处理，一旦连接触发了事件，就调用该函数，自己触发了什么事件如何处理自己决定
    {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if(_read_callback) _read_callback();
        }
        // 有可能会释放连接的操作一次只处理一个
        if(_revents & EPOLLOUT)
        {
            if(_write_callback) _write_callback();
        }
        else if(_revents & EPOLLERR)
        {
            if(_error_callback) _error_callback();
        }
        else if(_revents & EPOLLHUP)
        {
            if(_close_callback) _close_callback();
        }
        
        if(_event_callback) _event_callback();
    }
};

// 对epoll进行封装的模块
#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    unordered_map<int, Channel*> _channels;
private:
    // 对epoll的直接操作
    void Update(Channel* channel, int op)
    {

        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if(ret < 0)
            ERR_LOG("epoll_ctl failed");

    }
    // 判断一个channel是否已经添加了监控事件
    bool HasChannel(Channel* channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end())
            return false;
        return true;
    }
public:
    Poller() 
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if(_epfd < 0)
        {
            ERR_LOG("epoll create failed");
            abort();    //退出程序
        }
    }
    // 添加或修改监控事件
    void UpdateEvent(Channel* channel)
    {
        bool ret = HasChannel(channel);
        if(ret == false)
        {
            // 不存在则添加
            _channels.insert(make_pair(channel->Fd(), channel));
            return Update(channel, EPOLL_CTL_ADD);
        }

        return Update(channel, EPOLL_CTL_MOD);
        
    }
    // 移除监控
    void RemoveEvent(Channel* channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end())
            _channels.erase(it);
        Update(channel, EPOLL_CTL_DEL);
    }
    // 开始监控,返回活跃连接
    void poll(vector<Channel*>* active)
    {
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if(nfds < 0)
        {
            if(errno == EINTR)
                return;
            ERR_LOG("epoll wait err:%s",strerror(errno));
            abort();
        }
        for(int i = 0; i < nfds; ++i)
        {
            auto it = _channels.find(_evs[i].data.fd);
            if(it == _channels.end()) 
                return ;
            it->second->SetRevents(_evs[i].events); //设置实际就绪的事件
            active->push_back(it->second);
        }
    }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = function<void()>;
class TimerTask
{
private:
    uint64_t _id;           //定时器任务对象ID
    uint32_t _timeout;      //定时任务超时时间
    bool _canceled;         //定时任务取消标记，true 为取消定时任务，false 为不取消
    TaskFunc _task_cb;      //定时器对象要执行的定时任务
    ReleaseFunc _release;   //用于删除TimerWheel中保存的定时器对象信息
public:
    TimerTask(uint64_t id, uint32_t delay, const TaskFunc& cb)
        :_id(id), _timeout(delay), _task_cb(cb), _canceled(false)
        {}
    
    ~TimerTask() 
    {
        if(_canceled == false)
            _task_cb(); 
        
        _release();
    }
    
    void SetRelease(const ReleaseFunc& cb) { _release = cb; }   //设置删除定时器对象信息的 回调函数
    uint32_t DelayTime() { return _timeout; }                   //返回超时时间
    void cancel() { _canceled = true; }                         //将定时器标记为取消
};

class TimerWheel
{
private:
    using PtrTask = shared_ptr<TimerTask>;
    using WeakTack = weak_ptr<TimerTask>;
    int _tick;          //当前的秒针，走到哪释放哪里的任务
    int _capacity;      //表盘的最大数量 --其实就是最大延迟时间
    vector<vector<PtrTask>> _wheel;
    unordered_map<uint64_t, WeakTack> _timers;

    EventLoop* _loop;
    int _timerfd;   //定时器描述符---可读事件回调就是读取计数器，执行定时任务
    unique_ptr<Channel> _timer_channel;
private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it != _timers.end())
            { _timers.erase(it); }
    }
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if( timerfd < 0)
        {
            ERR_LOG("create timerfd failed");
            abort();
        }
        struct itimerspec itime;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;  //第一次超时时间为1秒后
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;     //第一次超时后，每次超时的时间
        timerfd_settime(timerfd, 0, &itime,nullptr);

        return timerfd;
    }
    int ReadTimeFd()
    {
        uint64_t times;
        // 有可能因为其他描述符的事件处理花费时间很长，然后在处理定时器事件描述符的时候，有可能就已经超时很多次了
        // read读取到的数据times就是从上一次read之后超时的次数
        int ret = read(_timerfd, &times, 8);
        if(ret < 0)
        {
            ERR_LOG("read timefd failed");
            abort();
        }
        return times;
    }
    // 这个函数应该每秒钟被执行一次，相当于秒针向后走了一步
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();      //清空指定位置的数组，就会把数组中保存的所有管理定时器对象的shared_ptr释放掉
    }
    void OnTime()
    {
        int times = ReadTimeFd();
        for(int i = 0; i< times; ++i)
            RunTimerTask();
    }
    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc& cb)      //添加定时任务
    {
        PtrTask pt(new TimerTask(id, delay, cb));
        pt->SetRelease(bind(&TimerWheel::RemoveTimer, this, id));
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);
        _timers[id] = WeakTack(pt);
    }
    void TimerRefreshInLoop(uint64_t id)                                     //刷新/延迟定时任务
    {
        // 通过保存的定时器对象的weak_ptr构造一个shared_ptr出来，添加到轮子中
        auto it = _timers.find(id);
        if(it == _timers.end())
            { return; } //没找到定时器任务，没办法刷新/延迟
        PtrTask pt = it->second.lock();         //lock获取weak_ptr管理的对象对应的shared_ptr
        int delay = pt->DelayTime();
        int pos = (_tick + delay ) % _capacity;
        _wheel[pos].push_back(pt);
    }
    void TimercancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            { return; } 
        PtrTask pt = it->second.lock();         //lock获取weak_ptr管理的对象对应的shared_ptr
        if(pt)
            pt->cancel();
    }
public:
    TimerWheel(EventLoop* loop)
        :_tick(0), _capacity(60), _wheel(_capacity)
        ,_loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd)) 
        {
            _timer_channel->SetReadCallback(bind(&TimerWheel::OnTime,this));
            _timer_channel->EnableRead();   //启动读事件监控
        }
    // 定时器中有个times成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题
    // 如果不想加锁，那就把所有对定时器的操作，都放到一个线程中执行
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb);
    // 刷新/延迟定时任务
    void TimerRefresh(uint64_t id);
    void Timercancel(uint64_t id);
    // 这个接口存在线程安全问题，这个接口不能被外界使用者调用，只能在模块内在对应的EventLoop线程内执行
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return false; 
        return true;
    }
};

// 对所有的描述符进行事件监控
class EventLoop
{
private:
    using Functor = function<void()>;   
    thread::id _thread_id;      //线程ID
    int _event_fd;      // 用于唤醒IO事件监控有可能导致的阻塞
    unique_ptr<Channel> _event_channel;
    Poller _poller;     // 进行所有描述符的事件监控
    vector<Functor> _tasks;     // 任务池
    mutex _mutex;               // 锁 用于任务池操作的线程安全 
    TimerWheel _timer_wheel;    //定时器模块
private:
    // 执行任务池中的任务
    void RunAllTask()
    {
        vector<Functor> functor;
        {
            unique_lock<mutex> lock(_mutex);
            _tasks.swap(functor);
        }
        for(auto& f : functor)
        {
            f();
        }
    }
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if(efd < 0)
        {
            ERR_LOG("create eventfd failed");
            abort();    //让程序异常退出
        }
        return efd;
    }
    void ReadEventFd()
    {   
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));
        if(ret < 0)
        {
            if(errno == EINTR || errno == EAGAIN)
                return ;
            ERR_LOG("read eventfd failed");
            abort();
        }
        return ;
    }
    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if(ret < 0)
        {
            if(errno == EINTR)
                return ;
            ERR_LOG("write eventfd failed");
            abort();
        }
        return ;
    }
public:
    EventLoop()
        :_thread_id(this_thread::get_id()), _event_fd(CreateEventFd()), 
        _event_channel(new Channel(this, _event_fd)),_timer_wheel(this)
    {
        // 给_event_fd添加可读事件回调函数，读取_event_fd事件通知次数
        _event_channel->SetReadCallback(bind(&EventLoop::ReadEventFd,this));
        // 启动_event_fd的读事件监控
        _event_channel->EnableRead();
    }
    // 分三步执行：事件监控 》就绪事件处理》执行任务
    void start()
    {
        while(true)
        {
            // 事件监控 
            vector<Channel*> actives;
            _poller.poll(&actives);
            // 就绪事件处理
            for(auto& channel: actives)
            {
                channel->HandlerEvent();
            }
            // 执行任务
            RunAllTask();
        }
    }
    // 用于判断当前线程是否是EventLoop对应的线程
    bool IsInLoop()
    {
        return (_thread_id == this_thread::get_id());
    }
    void AssertInLoop()
    {
        assert(_thread_id == this_thread::get_id());
    }
    // 判断将要执行的任务是否处于线程中，如果是则执行，不是则压入任务池
    void RunInLoop(const Functor& cb)
    {
        if(IsInLoop())
        {
            return cb();
        }
        return QueueInLoop(cb);
    }
    // 将操作压入任务池
    void QueueInLoop(const Functor& cb)
    {
        {
            unique_lock<mutex> lock(_mutex);
            _tasks.push_back(cb);

        }
        // 唤醒因为没有事件就绪，而导致的epoll阻塞
        // 其实就是给eventfd写入一个数据，eventfd就会触发可读事件
        WeakUpEventFd();
    }
    // 添加/修改描述符的事件监控
    void UpdateEvent(Channel* channel)
    {
        return _poller.UpdateEvent(channel);
    }
    // 移除描述符的监控
    void RemoveEvent(Channel* channel)
    {
        return _poller.RemoveEvent(channel);
    }
    // 添加定时任务
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb)
    {
        return _timer_wheel.TimerAdd(id, delay, cb);
    }
    //刷新定时任务
    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefresh(id);
    }
    // 取消定时任务
    void Timercancel(uint64_t id)
    {
        return _timer_wheel.Timercancel(id);
    }
    // 判断定时任务是否存在
    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }
};

class LoopThread
{
private:
    // 用于实现_loop获取同步关系，避免线程创建了，但是loop还没有实例化，如果获取loop就为空了
    mutex _mutex;               // 互斥锁
    condition_variable _cond;   // 条件变量
    EventLoop* _loop;       // EventLoop指针变量，这个对象需要在线程内实例化
    thread _thread;         // EventLoopd对应的线程
private:
    // 实例化EventLoop对象，并且开始运行EventLoop模块的功能
    void ThreadEntry()
    {
        EventLoop loop;
        {
            unique_lock<mutex> lock(_mutex);        //加锁
            _loop = &loop;
            _cond.notify_all();
        }
        loop.start();
    }
public:
    // 创建线程，设定线程入口函数 
    LoopThread():_loop(nullptr), _thread(thread(&LoopThread::ThreadEntry, this)) {}
    // 返回当前线程关联的EventLoop对象指针
    EventLoop* GetLoop()
    {
        EventLoop* loop = nullptr;
        {
            unique_lock<mutex> lock(_mutex);        //加锁
            _cond.wait(lock,[&](){ return _loop != nullptr; }); //条件变量，_loop为空就阻塞
            loop = _loop;
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    int _thread_count;
    int _next_idx;
    EventLoop* _base_loop;
    vector<LoopThread*> _threads;
    vector<EventLoop*> _loops;
public:
    LoopThreadPool(EventLoop* baseloop)
        :_thread_count(0),_next_idx(0), _base_loop(baseloop)
        {}
    void SetThread_count(int count) { _thread_count = count; }
    void CreateThreadPool()
    {
        if(_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i = 0; i < _thread_count; ++i)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    EventLoop* NextLoop()
    {
        if(_thread_count == 0)
            return _base_loop;
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};
// 通用的上下文类型
class Any
{
private:
    class holder
    {   
    public:
        virtual ~holder(){}
        virtual const std::type_info& type() = 0;
        virtual holder *clone() = 0;
    };
    template<class T>
    class placeholder :public holder
    {
    public:
        placeholder(const T& val):_val(val) {}
        // 获取子类对象保存的数据类型
        virtual const std::type_info& type() { return typeid(T); } 
        // 针对当前对象自身，克隆出一个新的子类对象
        virtual holder* clone() { return new placeholder(_val); }        
    public:
        T _val;
    };
    holder* _content;
public:
    Any():_content(nullptr) {}
    template<class T>
    Any(const T& val):_content(new placeholder<T>(val)) {}
    Any(const Any& other):_content(other._content ? other._content->clone() : nullptr) {}
    ~Any() { delete _content; }
    Any& Swap(Any& other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    //返回子类对象保存的数据的指针
    template<class T>
    T* get()       
    {
        // 想要获取的数据类型，必须和保存的数据类型一致
        assert(typeid(T) == _content->type());
        return &((placeholder<T>*)_content)->_val;
    }
    //赋值运算符重载函数，可以给子类对象设置数据类型 
    template<class T>
    Any& operator=(const T& val)
    {
        // 为val构造一个临时的通用容器，然后与当前容器自身进行指针交换
        // 临时对象释放的时候，临时对象释放的时候，原先保存的数据也就被释放了
        Any(val).Swap(*this);
        return *this;
    }    
    Any& operator=(const Any& other)
    {
        Any(other).Swap(*this);
        return *this;
    }
};

class Connection;
// DISCONNECTED：连接关闭状态， CONNECTING：连接建立成功--待处理状态
// CONNECTED：连接建立完成，各种设置已完成，可以通信的状态 DISCONNECTING：待关闭状态
typedef enum{DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING} Connstatu;
using PtrConnection = shared_ptr<Connection>;
class Connection :public enable_shared_from_this<Connection>
{
private:    
    uint64_t _next_id;      // 连接的唯一id，便于连接的管理和查找
    // uint64_t _timer_id;     // 定时器id，必须是唯一的，这里为了简化操作使用_conn_id作为定时器id
    int _sockfd;            // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动，非活跃销毁的判断标志，默认为false
    EventLoop* _loop;       // 连接所关联的一个EventLoop
    Connstatu _statu;       // 连接状态
    Socket _socket;         // 套接字操作管理
    Channel _channel;       // 连接的事件管理
    Buffer _in_buffer;      // 输入缓冲区 -- 存放从socket中读取到的数据
    Buffer _out_buffer;     // 输出缓冲区 -- 存放要发送给对端的数据
    Any _context;           // 请求接受处理的上下文
    // 这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）
    // 换句话说，这几个回调都是组件使用者设置的
    using ConnectedCallback = function<void(const PtrConnection&)>;
    using MessageCallback = function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = function<void(const PtrConnection&)>;
    using AnyEventCallback = function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    // 组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，
    // 一旦某个连接要关闭，就应该从管理的地方移除掉自己的信息
    ClosedCallback _server_closed_callback;

private:
    // 五个channel事件的回调函数 
    //描述符可读事件触发后调用的函数，接收socket数据放入接收缓冲区中，然后调用_message_callback
    void HandleRead()
    {
        // 1. 接收socket数据，放到缓冲区
        char buf[65536] = {0};
        int ret = _socket.NonBlockRecv(buf, sizeof(buf)-1);
        if(ret < 0)
            return ShutDownInLoop();    //出错了，不能直接关闭连接
        
        // 这里读取到0表示没有读取到数据，并不是连接断开了，连接断开返回的是-1
        _in_buffer.WriteAndPush(buf, ret);  //将数据放入输入缓冲区
        // 2. 调用 _message_callback进行业务处理、
        if(_in_buffer.ReadableSize() > 0)
            return _message_callback(shared_from_this(), &_in_buffer);
    }
    //描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
    void HandleWrite()
    {
        // _out_buffer中保存的数据就是待发送数据
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadableSize());
        if(ret < 0)
        {
            // 接受缓冲区中有数据就处理一下，然后释放连接
            if(_in_buffer.ReadableSize() > 0)
                _message_callback(shared_from_this(), &_in_buffer);
            return Release();    
        }
        // 要将读偏移向后移动，否则永远有数据
        _out_buffer.MoveReadOffset(ret);

        if( _out_buffer.ReadableSize() == 0)
        {
            _channel.DisableWrite();        //缓冲区中没有数据了，关闭写事件监控
        // 如果当前连接是待关闭状态 ，直接释放
            if(_statu == DISCONNECTING)
                return Release();   
        }

    }     
    void HandleClose()     //描述符触发挂断事件
    {
        // 一旦连接挂断了，套接字就什么都干不了了，因此有数据 就处理一下，完成后关闭连接
        if(_in_buffer.ReadableSize() > 0)
            _message_callback(shared_from_this(), &_in_buffer);
        return Release();
    }
    void HandleError()     //描述符触发错误时间
    {
        return HandleClose();
    }
    void HandleEvent()     //描述符触发任意事件
    {
        if(_enable_inactive_release == true)
            _loop->TimerRefresh(_next_id);
        if(_event_callback)
            _event_callback(shared_from_this());
    }
    //连接获取之后，所处的状态下要进行各种设置（启动读监控，调用回调函数）
    void EstablishedInLoop()   
    {
        // 1. 修改连接状态 2. 启动读事件监控 3. 调用回调函数
        assert(_statu == CONNECTING);   //当前的状态必须是上层的半连接状态
        _statu = CONNECTED;             //当前的函数执行完毕，则连接进入已完成连接状态
        // 一旦启动读事件监控可能会立即触发读事件，所以不能放在构造函数中启动
        _channel.EnableRead();
        if(_connected_callback)
            _connected_callback(shared_from_this());

    }
    void RealeaseInLoop()      //这个接口才是实际的释放接口
    {
        // 1. 修改连接状态，将其设置为DISCONNECTED
        _statu = DISCONNECTED;
        // 2. 移除连接的事件监控
        _channel.Remove();
        // 3. 关闭描述符
        _socket.Close();
        // 4. 如果当前定时器队列还有定时销毁任务,则取消任务
        if(_loop->HasTimer(_next_id)) CancelInactiveReleaseInLoop();
        // 5.调用关闭回调函数--先调用用户的关闭回调，在移除服务器管理的连接信息，否则会出错
        if(_closed_callback) _closed_callback(shared_from_this());
        // 移除服务器内部管理的连接信息
        if(_server_closed_callback) _server_closed_callback(shared_from_this());
    }

    // 这个并不是实际的发送接口，而是把数据放到发送缓冲区中，启动写事件监控
    void SendInLoop(Buffer& buf)
    {
        if(_statu == DISCONNECTED) return;
        _out_buffer.WriteBufferAndPush(buf);
        if(_channel.writeAble() == false)
            _channel.Enablewrite();
    }
    // 这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
    void ShutDownInLoop()
    {
        _statu = DISCONNECTING;     //设置连接为半关闭状态
        if(_in_buffer.ReadableSize() > 0)   //把接收缓冲区的数据处理一下
            if(_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        
        if(_out_buffer.ReadableSize() > 0)  //发送缓冲区中有数据，就启动写事件监控，就绪了就发送
            if(_channel.writeAble() == false)
                _channel.Enablewrite();
        if(_out_buffer.ReadableSize() == 0) //没有数据待发送，直接释放连接
            Release();
    }
    // 启动非活跃连接超时释放
    void EnableInactiveReleaseInLoop(int sec)
    {
        // 1. 将判断标志改为true
        _enable_inactive_release = true;
        // 2. 如果当前定时销毁任务已存在，则刷新一下
        if(_loop->HasTimer(_next_id))
            return _loop->TimerRefresh(_next_id);
        // 3. 如果不存在则新增定时销毁任务
        _loop->TimerAdd(_next_id, sec, bind(&Connection::Release, this));
    }
    // 取消非活跃连接超时释放
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if(_loop->HasTimer(_next_id))
            _loop->Timercancel(_next_id);
    }
    void UpgradeInLoop(const Any& context, const ConnectedCallback& conn, const MessageCallback& msg,
                 const ClosedCallback& closed, const AnyEventCallback& event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }
public:
    Connection(EventLoop* loop, uint64_t conn_id, int sockfd)
        : _next_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false)
        , _loop(loop), _statu(CONNECTING), _socket(_sockfd), _channel(_loop, _sockfd)
    {
        _channel.SetCloseCallback(bind(&Connection::HandleClose,this));
        _channel.SetReadCallback(bind(&Connection::HandleRead,this));
        _channel.SetWriteCallback(bind(&Connection::HandleWrite,this));
        _channel.SetErrorCallback(bind(&Connection::HandleError,this));
        _channel.SetEventCallback(bind(&Connection::HandleEvent,this));
    }
    ~Connection(){ DBG_LOG("release connection:%p",this); }
    int Fd(){ return _sockfd; }           //获取管理的文件描述符
    int Id(){ return _next_id; }           //获取连接ID
    bool Connected(){ return (_statu == CONNECTED); }   //是否处于connected状态
    // 设置上下文--连接建立完成时进行调用
    void SetContext(const Any& context) { _context = context; }
    // 获取上下文--返回的是指针
    Any* GetContext() { return &_context; }
    void SetConnectedCallback(const ConnectedCallback& cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback& cb){ _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback& cb){ _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback& cb){ _event_callback = cb; }
    void SetServerClosedCallback(const ClosedCallback& cb){ _server_closed_callback = cb; }

    // 连接建立就绪，进行_channel回调设置，启动读监控，调用ConnectedCallback
    void Established()
    {
        _loop->RunInLoop(bind(&Connection::EstablishedInLoop, this));
    }
    // 发送数据，将数据放到发送缓冲区，启动写事件监控
    void Send(const char* data, size_t len)
    {
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(bind(&Connection::SendInLoop, this, move(buf)));
    }
    // 提供给组件使用者的关闭接口，---并不实际关闭，需要判断有没有数据待处理
    void ShutDown()
    {
        _loop->RunInLoop(bind(&Connection::ShutDownInLoop, this));
    }
    void Release()
    {
        _loop->QueueInLoop(bind(&Connection::RealeaseInLoop,this));
    }
    // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    // 取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 切换协议---重置上下文以及阶段性回调处理函数---非线程安全
    // 这个接口必须在eventloop线程中立即执行，，为了防备新的事件触发后，处理的时候，切换任务还没被执行，--会导致数据使用原协议进行处理
    void Upgrade(const Any& context, const ConnectedCallback& conn, const MessageCallback& msg,
                 const ClosedCallback& closed, const AnyEventCallback& event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(bind(&Connection::UpgradeInLoop, this,context, conn, msg, closed, event));
    }
};

class Acceptor
{
private:
    Socket _socket;     // 用于创建监听套接字
    EventLoop* _loop;   // 用于对监听套接字进行事件监控
    Channel _channel;   // 用于对监听套接字进行事件管理

    using AcceptorCallback = function<void(int)>;
    AcceptorCallback _acceptor_callback;
private:
    // 监听套接字的读事件回调函数---获取新链接，调用_acceptor_callback函数进行新连接处理
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if(newfd < 0) return;
        if(_acceptor_callback) _acceptor_callback(newfd);
    }
    int CreateServer(uint16_t port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }
public:
    Acceptor(EventLoop* loop, uint16_t port)
        :_socket(CreateServer(port)), _loop(loop),_channel(loop, _socket.Fd())
    {
        _channel.SetReadCallback(bind(&Acceptor::HandleRead,this));
    }
    void SetAcceptorCallback(const AcceptorCallback& cb) { _acceptor_callback = cb; }
    void StartListen() { _channel.EnableRead(); }
};

class TcpServer
{
private:
    uint64_t _next_id;              //自增长的连接ID
    uint16_t _port;
    int _timeout;                   //非活跃连接的超时时间，多长时间无通信就是非活跃连接
    bool enable_inactive_release;   //是否启动了非活跃连接超时销毁的判断标志
    EventLoop _base_loop;           //主线程的EventLoop对象，负责监听事件的处理
    Acceptor _acceptor;             //监听套接字的管理对象
    LoopThreadPool _loop_pool;       //从属EventLoop的线程池
    unordered_map<uint64_t, PtrConnection> _conns;  //保存所有连接对应的shared_ptr对象
    using Functor = function<void()>;
    using ConnectedCallback = function<void(const PtrConnection&)>;
    using MessageCallback = function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = function<void(const PtrConnection&)>;
    using AnyEventCallback = function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
private:
    void RunAfterInLoop(const Functor& task, int delay)
    {
        _next_id++;
        _base_loop.TimerAdd(_next_id, delay, task);
    }
    // 为新链接构造一个Connection进行管理
    void NewConnection(int fd)     
    {
        _next_id++;
        PtrConnection conn(new Connection(_loop_pool.NextLoop(), _next_id, fd));
        conn->SetConnectedCallback(_connected_callback);
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetServerClosedCallback(bind(&TcpServer::RemoveConnection, this, placeholders::_1));
        if(enable_inactive_release) conn->EnableInactiveRelease(_timeout);    // 启动非活跃超时销毁
        conn->Established();                // 就绪初始化 // 1. 修改连接状态 2. 启动读事件监控 3. 调用回调函数
        _conns.insert(make_pair(_next_id, conn));

    }
    void RemoveConnectionInLoop(const PtrConnection& conn)
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if(it != _conns.end())
        {
            _conns.erase(it);
        }
    }
    // 从管理Connection的conns中移除连接信息
    void RemoveConnection(const PtrConnection& conn)
    {
        _base_loop.RunInLoop(bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }        
public:
    TcpServer(uint16_t port)
        : _next_id(0), _port(port),_timeout(0), enable_inactive_release(false)
        , _acceptor(&_base_loop, _port), _loop_pool(&_base_loop)
        {
            _acceptor.SetAcceptorCallback(bind(&TcpServer::NewConnection,this,placeholders::_1));
            _acceptor.StartListen();        //将监听套接字挂到baseloop上
        }

    void SetThreadCount(int count) { return _loop_pool.SetThread_count(count); }
    void SetConnectedCallback(const ConnectedCallback& cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback& cb){ _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback& cb){ _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback& cb){ _event_callback = cb; }
    void EnableInactiveRelease(int timeout) { _timeout = timeout; enable_inactive_release = true; }
    void RunAfter(const Functor& task, int delay)    //用于添加一个定时任务
        { _base_loop.RunInLoop(bind(&TcpServer::RunAfterInLoop, this, task, delay)); }
    void StartServer() 
    {
        _loop_pool.CreateThreadPool();  //创建线程池中的从属线程,这个函数只能等其他的函数设置好了才能调用，否则会报错
        _base_loop.start(); 
    }

};

void Channel::Remove() { return _loop->RemoveEvent(this); }
void Channel::Update() { return _loop->UpdateEvent(this); }

void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb)
{
    _loop->RunInLoop(bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
}
void TimerWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(bind(&TimerWheel::TimerRefreshInLoop, this, id));
}
void TimerWheel::Timercancel(uint64_t id)
{
    _loop->RunInLoop(bind(&TimerWheel::TimercancelInLoop, this, id));
}


class NetWork
{
public:
    NetWork()
    {
        DBG_LOG("signal init");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;

#endif 